package com.itcast.spark

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, MapWithStateDStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, State, StateSpec, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * DESC:当前的案例主要是接受的Socket的数据，需要指明主机名和端口号接受数据
 * 注意在这里SparkStreaming采取的方式是receiver进程拉取数据源的数据，进行不同阶段的处理
 */
object SparkStremingTCPMapWithState {
  def main(args: Array[String]): Unit = {
    //1-环境准备
    val conf: SparkConf = new SparkConf().setAppName("SparkStremingTCP").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    //这里就是指定配置项将数据按照5秒为周期进行处理
    val ssc = new StreamingContext(sc, Seconds(5))
    ssc.checkpoint("./datasets/checkpoint/ck4")
    //2-准备socket
    val receiveData: ReceiverInputDStream[String] = ssc.socketTextStream("node01", 9999, StorageLevel.MEMORY_AND_DISK_2)
    //3-接下来对该数据进行简单的wordcount的统计
    val flatMapDS: DStream[String] = receiveData.flatMap(_.split(" "))
    val mapDS: DStream[(String, Int)] = flatMapDS.map(x => (x, 1))
    val reduceDS: MapWithStateDStream[String, Int, Int, Any] = mapDS.mapWithState(StateSpec.function(spec).timeout(Seconds(30)))
    //4-如果想对当前的结果进行排序的化，请问如何实现-transform(func)
    //5-执行结果的输出
    reduceDS.print()
    //6-开启程序执行---Start the execution of the streams.开启streams的执行
    ssc.start()
    //7-等到程序停止---等待执行器的停止，用户或程序触发任何异常都会引起当前程序停止
    ssc.awaitTermination()
    //8-停止streamingcontext
    //第一个参数就是停止saprkcontext
    //第二个参数就是优雅停止，等到所有的接收器接受的数据处理完毕之后再停止
    ssc.stop(true, true)

  }

  //分组聚合
  /**
   * 定义一个函数，该函数有三个类型word: String, option: Option[Int], state: State[Int]
   * 其中word代表统计的单词
   * option代表的是历史数据（使用option是因为历史数据可能有，也可能没有，如第一次进来的数据就没有历史记录），
   * state代表的是返回的状态,state的状态需要进行更新
   */

  val spec = (words: String, option: Option[Int], state: State[Int]) => {
    if (state.isTimingOut()) {
      //如果当前的状态设置上timeout的超时的时间，这里如果超时了直接打印警告信息
      println(words + "words state is time out:")
    } else {
      //首先将历史的数据进行处理---如果有就进行累加，如果没有赋值为0
      val sum: Int = option.getOrElse(0) + state.getOption().getOrElse(0)
      val resultTuple: (String, Int) = (words, sum)
      //根据state.update()进行状态的更新
      state.update(sum)
      resultTuple //返回当前改变的word的统计计数
    }
    // Use state.exists(), state.get(), state.update() and state.remove()
    // to manage state, and return the necessary string
  }
}
